package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"6.824/labgob"
	"6.824/utils"
	"bytes"
	"log"
	"time"

	//	"bytes"
	"sync"
	"sync/atomic"

	//	"6.824/labgob"
	"6.824/labrpc"
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
	CommandTerm  int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex // Lock to protect shared access to this peer's state
	cond      *sync.Cond
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()
	applyCh   chan ApplyMsg
	// Your data here (2A, 2B, 2C).
	persistentContext PersistentContext
	commitIndex       EntryIndex
	lastApplied       EntryIndex
	currentState      State
	timeoutStamp      int64
	clockMu           sync.Mutex

	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	follower  *smfollower
	leader    *smleader
	candidate *smcandidate

	needPersist       bool
	snapshotToPersist []byte
}

func (rf *Raft) begin() {
	rf.mu.Lock()
	utils.Assert(rf.needPersist == false)
}
func (rf *Raft) end() {
	if rf.needPersist {
		rf.persist()
	}
	rf.needPersist = false
	rf.mu.Unlock()
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	rf.begin()
	defer rf.end()
	term = rf.persistentContext.CurrentTerm
	isleader = rf.currentState == LeaderState
	// Your code here (2A).
	return term, isleader
}

func (rf *Raft) serializeCtx() []byte {
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	if e.Encode(rf.persistentContext) != nil {
		log.Fatalf("fail to encode")
	}
	return w.Bytes()
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {

	if rf.snapshotToPersist != nil {
		rf.persister.SaveStateAndSnapshot(rf.serializeCtx(), rf.snapshotToPersist)
		rf.snapshotToPersist = nil
	} else {
		rf.persister.SaveRaftState(rf.serializeCtx())
	}
	/*
		if e.Encode(rf.persistentContext.CurrentTerm) != nil || e.Encode(rf.persistentContext.VotedFor) != nil || e.Encode(rf.persistentContext.LastLogIndex) != nil {
			log.Fatalf("Fail to encode")
		}
		if e.Encode(rf.persistentContext.SnapshotIndex) != nil || e.Encode(rf.persistentContext.SnapshotTerm) != nil {
			log.Fatalf("Fail to encode")
		}

		localIndex := rf.persistentContext.GetLastLocalIndex()
		for i := 0; i <= localIndex; i++ {
			if e.Encode(rf.persistentContext.LogEntries[i]) != nil {
				log.Fatalf("Fail to encode")
			}
		}*/

	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)

	rf.persistentContext = PersistentContext{}

	if d.Decode(&rf.persistentContext) != nil {
		log.Fatalf("fail t decode")
	}

	/*
		var t, v int
		var i EntryIndex

		if d.Decode(&t) != nil || d.Decode(&v) != nil || d.Decode(&i) != nil {
			log.Fatalf("Fail to decode #0")
		}

		rf.persistentContext.CurrentTerm = t
		rf.persistentContext.VotedFor = v
		rf.persistentContext.LastLogIndex = i

		var st int
		var si EntryIndex
		if d.Decode(&si) != nil || d.Decode(&st) != nil {
			log.Fatalf("Fail to decode #1")
		}
		rf.persistentContext.SnapshotIndex = si
		rf.persistentContext.SnapshotTerm = st

		rf.persistentContext.Snapshot = rf.persister.ReadSnapshot()

		lst := rf.persistentContext.GetLastLocalIndex()
		rf.persistentContext.LogEntries = make([]LogEntry, 0, lst+1)
		for idx := 0; idx <= lst; idx++ {
			var entry LogEntry
			if d.Decode(&entry) != nil {
				log.Fatalf("Fail to decode #2")
			}
			rf.persistentContext.LogEntries = append(rf.persistentContext.LogEntries, entry)
		}
	*/

	rf.commitIndex = Max(rf.commitIndex, rf.persistentContext.SnapshotIndex)

	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {

	// Your code here (2D).
	rf.begin()
	defer rf.end()
	if EntryIndex(index) <= rf.persistentContext.SnapshotIndex {
		return
	}

	rf.Log("SNAP", "start snapshot")
	//log.Printf("index:%v len:%v snp index:%v", index, len(rf.persistentContext.LogEntries), rf.persistentContext.SnapshotIndex)
	utils.Assert(rf.persistentContext.ToLocalIndex(EntryIndex(index)) < len(rf.persistentContext.LogEntries))
	utils.Assert(index <= int(rf.lastApplied))

	//log.Printf("snp index:%v lstapply:%v lstsnapindex:%v", index, rf.lastApplied, rf.persistentContext.SnapshotIndex)
	tmp := rf.persistentContext.LogEntries[rf.persistentContext.ToLocalIndex(EntryIndex(index)+1) : rf.persistentContext.GetLastLocalIndex()+1]

	rf.persistentContext.SnapshotTerm = rf.persistentContext.GetEntryTerm(EntryIndex(index))
	rf.persistentContext.SnapshotIndex = EntryIndex(index)

	rf.persistentContext.LogEntries = make([]LogEntry, len(tmp))
	copy(rf.persistentContext.LogEntries, tmp)
	rf.needPersist = true
	rf.snapshotToPersist = snapshot

}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// Term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	if command == nil {
		log.Fatalf("cmd is nil")
	}

	rf.begin()
	if rf.currentState == LeaderState {
		i := rf.persistentContext.GetLastIndex() + 1
		t := rf.persistentContext.CurrentTerm
		rf.appendLog(LogEntry{t, command}, i)

		rf.Log("CLNT", "client start, wait for index:%v", i)

		rf.leader.sendNewEntries()
		rf.end()
		index = int(i)
		term = t
		//rf.LogFree("CLNT", "client finish, index:%v, term:%v", index, term)

	} else {
		rf.end()
		isLeader = false

	}
	//	rf.Log("CLNT", "index:%d, term:%d, isLeader:%v", index, term, isLeader)

	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

// The ticker go routine starts a new election if this peer hasn't received
// heartsbeats recently.
func (rf *Raft) ticker() {
	for rf.killed() == false {

		now := time.Now().UnixMilli()
		rf.clockMu.Lock()
		if rf.timeoutStamp-10 < now {

			rf.clockMu.Unlock()

			rf.begin()
			rf.Log("TIMR", "time out")

			if rf.currentState == FollowerState || rf.currentState == CandidateState {

				rf.changeToState(CandidateState, rf.persistentContext.CurrentTerm+1)
			}
			rf.end()

			rf.resetTimeoutClock()
		} else {

			rf.clockMu.Unlock()
		}

		time.Sleep(time.Duration(HeartBeatsTime) * time.Millisecond)

		// Your code here to check if a leader election should
		// be started and to randomize sleeping time using
		// time.Sleep().

	}
}
func (rf *Raft) applyLoop() {

	for rf.killed() == false {
		rf.cond.L.Lock()
		for rf.commitIndex == rf.lastApplied {
			rf.cond.Wait()
		}

		utils.Assert(rf.commitIndex > rf.lastApplied)
		utils.Assert(rf.commitIndex <= rf.persistentContext.GetLastIndex())
		var msg ApplyMsg
		if rf.lastApplied < rf.persistentContext.SnapshotIndex {
			msg = ApplyMsg{CommandValid: false, SnapshotValid: true, Snapshot: rf.persister.ReadSnapshot(), SnapshotIndex: int(rf.persistentContext.SnapshotIndex), SnapshotTerm: rf.persistentContext.SnapshotTerm}
			rf.lastApplied = rf.persistentContext.SnapshotIndex
			rf.Log("LOG1", "apply snapshot, index:%v term:%v", rf.persistentContext.SnapshotIndex, rf.persistentContext.SnapshotTerm)
		} else {
			cmd := rf.persistentContext.GetEntry(rf.lastApplied + 1)
			msg = ApplyMsg{CommandValid: true, CommandIndex: int(rf.lastApplied + 1), CommandTerm: rf.persistentContext.CurrentTerm, Command: cmd.Cmd}
			if val, b := cmd.Cmd.(int); b {

				rf.Log("LOG1", "apply index:%d, val: %v", rf.lastApplied+1, val)
			} else {

				rf.Log("LOG1", "apply index:%d", rf.lastApplied+1)
			}
			rf.lastApplied++
		}

		rf.cond.L.Unlock()

		rf.applyCh <- msg

	}

}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.cond = sync.NewCond(&rf.mu)
	rf.applyCh = applyCh

	rf.persistentContext = PersistentContext{1, -1, make([]LogEntry, 1), -1, -1} //term initial is 1 because use negative return val to identify others term
	rf.persistentContext.LogEntries[0].Term = 1
	rf.currentState = FollowerState
	rf.lastApplied = 0
	rf.commitIndex = 0

	rf.candidate = &smcandidate{rf}
	rf.leader = &smleader{rf, nil, nil, 0}
	rf.follower = &smfollower{rf}

	rf.needPersist = false
	rf.snapshotToPersist = nil

	rf.resetTimeoutClock()
	// Your initialization code here (2A, 2B, 2C).

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	rf.Log("INFO", "start server")
	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.applyLoop()

	return rf
}
